package com.yisoo.ubag.iot.component;


import com.aliyuncs.IAcsClient;
import com.rabbitmq.tools.json.JSONUtil;
import com.yisoo.common.to.EmailDataTo;
import com.yisoo.common.utils.R;
import com.yisoo.ubag.iot.entity.MsgBackEntity;
import com.yisoo.ubag.iot.manager.AmqpManager;
import org.apache.commons.codec.binary.Base64;
import org.apache.qpid.jms.JmsConnection;
import org.apache.qpid.jms.JmsConnectionListener;
import org.apache.qpid.jms.message.JmsInboundMessageDispatch;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.web.context.request.RequestContextHolder;

import javax.crypto.Mac;
import javax.crypto.spec.SecretKeySpec;
import javax.jms.*;
import javax.naming.Context;
import javax.naming.InitialContext;
import java.net.URI;
import java.util.Hashtable;
import java.util.UUID;
import java.util.concurrent.*;

@Component
public class AmqpComponent {

    @Autowired
    ThreadPoolExecutor executorService;

    @Autowired
    AmqpManager amqpManager;

    @Autowired
    public StringRedisTemplate stringRedisTemplate;

    @Autowired
    RabbitTemplate rabbitTemplate;

    private final static Logger logger = LoggerFactory.getLogger(AmqpComponent.class);


    /**
     * 在这里处理您收到消息后的具体业务逻辑。
     */
    private void processMessage(Message message) {
        try {
            //TODO 调试时使用
            if (message == null){
                System.out.println("message为空");
                return;
            }
            //在这里处理消费组所有的数据
            //1.先过一遍消息处理器：解析出消息对象
            MsgBackEntity msgBackEntity = amqpManager.parseAmqpDataToMsgBackEntity(message);
            //2.异步编排两个线程： 备份线程 和 处理线程
            System.out.println("平台服务端订阅："+msgBackEntity);
            CompletableFuture<Void> Future = CompletableFuture.runAsync(() -> {
                //消息备份，消息队列接收后直接会存入数据库，不做其他处理
                rabbitTemplate.convertAndSend("ubag.exchange.pub","mqtt.pub",msgBackEntity,new CorrelationData(UUID.randomUUID().toString()));
            }, executorService);
            CompletableFuture<Void> Future2 = CompletableFuture.runAsync(() ->{
                //消息处理，在process中分类
                amqpManager.runProcessor(msgBackEntity);
            }, executorService);
            try {
                CompletableFuture.allOf(Future,Future2).get();
            } catch (InterruptedException | ExecutionException e) {
                e.printStackTrace();
            }
        } catch (Exception e) {
            logger.error("processMessage occurs error ", e);
        }
    }

    //业务处理异步线程池，线程池参数可以根据您的业务特点调整，或者您也可以用其他异步方式处理接收到的消息。
//    private final static ExecutorService executorService = new ThreadPoolExecutor(
//            Runtime.getRuntime().availableProcessors(),
//            Runtime.getRuntime().availableProcessors() * 2, 60, TimeUnit.SECONDS,
//            new LinkedBlockingQueue<>(50000));

    @Value("${mqtt.user.accessKeyID}")
    private  String accessKey;

    @Value("${mqtt.user.accessKeySecret}")
    private  String accessSecret;

    @Value("${mqtt.iot.regionId}")
    private  String regionId;

    @Value("${mqtt.iot.consumerGroupId}")
    private  String consumerGroupId;

    @Value("${mqtt.iot.signMethod}")
    private  String signMethod;

    @Value("${mqtt.iot.aliyunId}")
    private  String aliyunId;

    private  String iotInstanceId = "";

    @Bean
    public MessageConsumer getMessageConsumer()  throws Exception {

        long timeStamp = System.currentTimeMillis();
        String clientId = UUID.randomUUID().toString();

        String userName = clientId + "|authMode=aksign"
                + ",signMethod=" + signMethod
                + ",timestamp=" + timeStamp
                + ",authId=" + accessKey
                + ",iotInstanceId=" + iotInstanceId
                + ",consumerGroupId=" + consumerGroupId
                + "|";

        String signContent = "authId=" + accessKey + "&timestamp=" + timeStamp;

        String password = doSign(signContent,accessSecret, signMethod);

        //接入域名，请参见AMQP客户端接入说明文档。
        String connectionUrl = "failover:(amqps://"+aliyunId+".iot-amqp."+regionId+".aliyuncs.com:5671?amqp.idleTimeout=80000)"
                + "?failover.reconnectDelay=30";

        Hashtable<String, String> hashtable = new Hashtable<>();
        hashtable.put("connectionfactory.SBCF",connectionUrl);
        hashtable.put("queue.QUEUE", "default");
        hashtable.put(Context.INITIAL_CONTEXT_FACTORY, "org.apache.qpid.jms.jndi.JmsInitialContextFactory");
        Context context = new InitialContext(hashtable);
        ConnectionFactory cf = (ConnectionFactory)context.lookup("SBCF");
        Destination queue = (Destination)context.lookup("QUEUE");


        // 创建连接。
        Connection connection = cf.createConnection(userName, password);

        JmsConnectionListener myJmsConnectionListener = new JmsConnectionListener() {
            //连接成功建立。
            @Override
            public void onConnectionEstablished(URI remoteURI) { logger.info("onConnectionEstablished, remoteUri:{}", remoteURI); }
            //尝试过最大重试次数之后，最终连接失败。
            @Override
            public void onConnectionFailure(Throwable error) { logger.error("onConnectionFailure, {}", error.getMessage()); }
            //连接中断。
            @Override
            public void onConnectionInterrupted(URI remoteURI) {logger.info("onConnectionInterrupted, remoteUri:{}", remoteURI); }
            // 连接中断后又自动重连上。
            @Override
            public void onConnectionRestored(URI remoteURI) { logger.info("onConnectionRestored, remoteUri:{}", remoteURI); }
            @Override
            public void onInboundMessage(JmsInboundMessageDispatch envelope) {}
            @Override
            public void onSessionClosed(Session session, Throwable cause) {}
            @Override
            public void onConsumerClosed(MessageConsumer consumer, Throwable cause) {}
            @Override
            public void onProducerClosed(MessageProducer producer, Throwable cause) {}
        };

        ((JmsConnection) connection).addConnectionListener(myJmsConnectionListener);


        // 创建会话。
        // Session.CLIENT_ACKNOWLEDGE: 收到消息后，需要手动调用message.acknowledge()。
        // Session.AUTO_ACKNOWLEDGE: SDK自动ACK（推荐）。
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        connection.start();


        // 创建Receiver连接。
        MessageConsumer consumer = session.createConsumer(queue);


        MessageListener messageListener = message -> {
            try {
                //1.收到消息之后一定要ACK。
                // 推荐做法：创建Session选择Session.AUTO_ACKNOWLEDGE，这里会自动ACK。
                // 其他做法：创建Session选择Session.CLIENT_ACKNOWLEDGE，这里一定要调message.acknowledge()来ACK。
                // message.acknowledge();
                //2.建议异步处理收到的消息，确保onMessage函数里没有耗时逻辑。
                // 如果业务处理耗时过程过长阻塞住线程，可能会影响SDK收到消息后的正常回调。
                executorService.submit(() -> processMessage(message));
            } catch (Exception e) {
                logger.error("submit task occurs exception ", e);
            }
        };


        consumer.setMessageListener(messageListener);
        System.out.println("jms监听启动完成");
        return consumer;
    }

    /**
     * 计算签名，password组装方法，请参见AMQP客户端接入说明文档。
     */
    private static String doSign(String toSignString, String secret, String signMethod) throws Exception {
        SecretKeySpec signingKey = new SecretKeySpec(secret.getBytes(), signMethod);
        Mac mac = Mac.getInstance(signMethod);
        mac.init(signingKey);
        byte[] rawHmac = mac.doFinal(toSignString.getBytes());
        return Base64.encodeBase64String(rawHmac);
    }
}
